/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.core;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import lac.cnet.sddl.objects.ApplicationObject;
import lac.cnet.sddl.objects.Message;
import lac.cnet.sddl.objects.MetaMessage;
import lac.cnet.sddl.objects.SpatialMessage;
import lac.cnet.sddl.udi.core.SddlLayer;
import lac.cnet.sddl.udi.core.UniversalDDSLayerFactory;
import lac.cnet.sddl.udi.core.UniversalDDSLayerFactory.SupportedDDSVendors;
import lac.cnet.sddl.udi.core.listener.UDIDataReaderListener;

/**
 * A bridge that links the API to receive/send data to/from the SDDL communication middleware with the CEP platform.
 *
 * This class bridge (links) the SDDL communication API to our CEP platform using a combination of the the Bridge
 * and Singleton pattern, which allow us to have a unique entry point to receive and send data in the platform.
 * 
 * Default Topics <Name, Type>
 * - <META, Message>
 * - <SPATIAL, Spatial>
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class DDSLink {
    
    /** LOGGER **/    
    private static Logger LOGGER = Logger.getLogger(DDSLink.class.getCanonicalName());

	// Singleton Instance
	protected static DDSLink instance;
	
    // JSON Mapper
	protected ObjectMapper mapper;
	
	// SDDL Layer Communication
	private SddlLayer core;
	
	// Instance name
	private String deployName;
	
	// Topic Lists
	private List<String> pubTopics;
	private List<String> subTopics;
    private Map<String, Object> topics;
	
	////////////////////////////////////////////////////////////////////////////////////////////////////
	// Singleton
	////////////////////////////////////////////////////////////////////////////////////////////////////
	
    // Default constructor
	protected DDSLink(SupportedDDSVendors ddsVendor) {
	    // DDS Core
	    core = UniversalDDSLayerFactory.getInstance(ddsVendor);
		core.createParticipant(UniversalDDSLayerFactory.CNET_DOMAIN);
		core.createPublisher();
		core.createSubscriber();

		// Topic Lists
		pubTopics = new ArrayList<>();
		subTopics = new ArrayList<>();
		topics = new HashMap<>();
		
		// JSON Mapper
		mapper = new ObjectMapper();
	}
	
	/** 
	 * DDS data stream link singleton.
	 * @param ddsVendor a specific dds vendor.
	 * @return a DDS link to receive and send data.
	 */
	public synchronized static DDSLink getInstance(SupportedDDSVendors ddsVendor) {
		if (instance == null) 
			instance = new DDSLink(ddsVendor); // TODO: Double locking

		return instance;
	}
	
	/**
	 * DDS data stream link singleton using tthe standard vendor (OpenSplice).
	 * 
	 * @return a DDS link to receive and send data.
	 */
	public static DDSLink getInstance() {
		return getInstance(SupportedDDSVendors.OpenSplice);
	}
	
	////////////////////////////////////////////////////////////////////////////////////////////////////
	// Pub/Sub Topics
	////////////////////////////////////////////////////////////////////////////////////////////////////
	
	public void createDefaultTopics() {
	    createMetaTopic("Meta");
	}
	
    /**
     * Create a publish topic with name topicName based on the {@link Message} class.
     * @param topicName the topic name
     */
    public void createMessageTopic(String topicName) {
        if (!topics.containsKey(topicName)) {
            LOGGER.info("Creating topic: " + topicName + " of type Message");
            topics.put(topicName, core.createTopic(Message.class, topicName));

            Object topic = topics.get(topicName);
            core.createDataWriter(topic);
            
            // Register
            pubTopics.add(topicName);
            subTopics.add(topicName);
        } else {
            LOGGER.error("Unable to create topic: " + topicName + " of type Message (already created)");
        }
    }
	
    /**
     * Create a publish topic with name topicName based on the {@link MetaMessage} class.
     * @param topicName the topic name
     */
    public void createMetaTopic(String topicName) {
        if (!topics.containsKey(topicName)) {
            LOGGER.info("Creating topic: " + topicName + " of type MetaMessage");
            topics.put(topicName, core.createTopic(MetaMessage.class, topicName));

            Object topic = topics.get(topicName);
            core.createDataWriter(topic);
            
            // Register
            pubTopics.add(topicName);
            subTopics.add(topicName);
        } else {
            LOGGER.error("Unable to create topic: " + topicName + " of type MetaMessage (already created)");
        }
    }
    
    /**
     * Create a publish topic with name topicName based on the {@link SpatialMessage} class.
     * @param topicName the topic name
     */
    public void createSpatialTopic(String topicName) {
        if (!topics.containsKey(topicName)) {
            LOGGER.info("Creating topic: " + topicName + " of type SpatialMessage");
            topics.put(topicName, core.createTopic(SpatialMessage.class, topicName));
            
            Object topic = topics.get(topicName);
            core.createDataWriter(topic);
            
            // Register
            pubTopics.add(topicName);
            subTopics.add(topicName);
        } else {
            LOGGER.error("Unable to create topic: " + topicName + " of type SpatialMessage (already created)");
        }
    }

    // Get and Sets
    public List<String> getPublishTopics() {
        return pubTopics;
    }
    
    public List<String> getSubscribeTopics() {
        return subTopics;
    }

    ////////////////////////////////////////////////////////////////////////////////////////////////////
    // Bridge Subscribe Methods
    ////////////////////////////////////////////////////////////////////////////////////////////////////
    
    /**
     * Adds a SDDL Listener to a given DDS topic.
     * 
     * @param topicName the topic name
     * @param listener who will be called when data is received.
     * 
     * @throws IllegalArgumentException when topic does not exist
     */
    public void addListener(String topicName, UDIDataReaderListener<ApplicationObject> listener) {
        if (topics.containsKey(topicName)) {
            LOGGER.info("Creating listener for topic: " + topicName);
            core.createDataReader(listener, topics.get(topicName));
        } else {
            throw new IllegalArgumentException("Topic " + topicName + " does not exist.");
        }
    }

    /**
     * Adds a filtered SDDL Listener to a given DDS topic.
     * 
     * @param topicName the topic name
     * @param listener who will be called when data is received.
     * @param filter logical string with filter (e.g., id = %0 and temp > %1)
     * @param filterParameters filter parameters
     * 
     * @throws IllegalArgumentException when topic does not exist
     */
    public void addFilteredListener(String topicName, UDIDataReaderListener<ApplicationObject> listener, 
                                    String filter, List<String> filterParameters) {
        LOGGER.info("Creating filtered listener for topic: " + topicName);
        if (topics.containsKey(topicName)) {
            Object filteredTopic = core.createContentFilteredTopic(topicName,
                                                                   topics.get(topicName), 
                                                                   filter, 
                                                                   filterParameters);
            core.createDataReader(listener, filteredTopic);
        } else {
            throw new IllegalArgumentException("Topic " + topicName + " does not exist.");
        }
    }
	
	////////////////////////////////////////////////////////////////////////////////////////////////////
	// Bridge Publish Methods
	////////////////////////////////////////////////////////////////////////////////////////////////////

    /**
     * Sends a message to a given topic (including metadata).
     * 
     * @param topic the topic name
     * @param obj the {@link Serializable} object to send
     * @param metadata a list of tags (TODO: key, value map?)
     */
    public void publishSpatial(String topic, double latitude, double longitude, String msgType, Serializable obj, String... metadata) {
        LOGGER.info("Writting " + obj + " to topic:" + topic);

        SpatialMessage spatialMSG = new SpatialMessage();
        
        try {
            spatialMSG.setEpnName(getDeployName());
            spatialMSG.setLatitude(latitude);
            spatialMSG.setLongitude(longitude);
            spatialMSG.setMsgType(msgType);
            spatialMSG.setContent(mapper.writeValueAsBytes(obj));
            spatialMSG.setStamps(Arrays.asList(metadata));

            core.writeTopic(topic, spatialMSG);
        } catch (JsonProcessingException ex) {
            LOGGER.error("Error at serializaing object to JSON");
        }
        LOGGER.info("Writting " + obj + " to topic:" + topic);
    }
    
    /**
     * Sends a message to a given topic (including metadata).
     * 
     * @param topic the topic name
     * @param obj the {@link Serializable} object to send
     * @param metadata a list of tags (TODO: key, value map?)
     */
    public void publishMSG(String topic, Serializable obj, String... metadata) {
        LOGGER.info("Writting " + obj + " to topic:" + topic);
        Message msg = new Message();
        try {
            msg.setContent(mapper.writeValueAsBytes(obj));
            msg.setStamps(Arrays.asList(metadata));
            core.writeTopic(topic, msg);
        } catch (JsonProcessingException ex) {
            LOGGER.error("Error at serializaing object to JSON");
        }

        LOGGER.info("Wrote " + obj + " to topic:" + topic);
    }

    /**
     * Sends a message to a given topic (including metadata).
     * 
     * @param topic the topic name
     * @param obj the {@link Serializable} object to send
     * @param metadata a list of tags (TODO: key, value map?)
     */
    public void publishMeta(String topic, String to, String command) {
        LOGGER.info("Writting " + command + " to topic:" + topic);
        MetaMessage msg = new MetaMessage();
        
        msg.setFrom(getDeployName());
        msg.setTo(to);
        msg.setCommand(command);

        core.writeTopic(topic, msg);
        LOGGER.info("Wrote " + command + " to topic:" + topic);
    }
    
	/**
     * Sends a message to a given topic (including metadata).
     * 
     * @param topic the topic name
     * @param obj the {@link Serializable} object to send
     * @param metadata a list of tags (TODO: key, value map?)
     */
    public void publish(String topic, Serializable obj, String... metadata) {
        LOGGER.info("Writting " + obj + " to topic:" + topic);
        MetaMessage msg = new MetaMessage();
        
        msg.setFrom(getDeployName());
        msg.setTo("ALL");
        msg.setCommand((String) obj);

        core.writeTopic(topic, msg);
        LOGGER.info("Wrote " + obj + " to topic:" + topic);
    }
	
	/**
	 * Sends a message to Outside 'topic' (Out)
     * 
     * @param obj the {@link Serializable} object to send
     * @param metadata a list of tags (TODO: key, value map?)
	 */
	public void publishOut(Serializable obj, String... metadata) {
	    publish("Out", obj, metadata);
	}
	
	/**
     * Sends a message to all publish 'topics'
     * @param optional metadata (tag)
     * @param payload
     */
    public void publishToAll(Serializable obj, String... metadata) {
        for (String topic : pubTopics) 
            publish(topic, obj, metadata);
    }
    
    ////////////////////////////////////////////////////////////////////////////////////////////////////
    // Close
    ////////////////////////////////////////////////////////////////////////////////////////////////////

    public void close() {
        core.deleteParticipant();
        core.deletePublisher();
        core.deleteSubscriber();
        pubTopics.clear();
        subTopics.clear();
        
        instance = null;
    }
    
    ////////////////////////////////////////////////////////////////////////////////////////////////////
    // Set and Gets
    ////////////////////////////////////////////////////////////////////////////////////////////////////
    
    public String getDeployName() {
        return deployName;
    }
    
    public void setDeployName(String deployName) {
        this.deployName = deployName;
    }
}
